package com.etl.service.impl;

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.etl.entity.EtlTaskIncrement;
import com.etl.enums.ResultEnum;
import com.etl.mapper.EtlTaskIncrementMapper;
import com.etl.util.AESUtil;
import com.etl.util.CommonUtil;
import com.etl.constants.CommonConstant;
import com.etl.constants.UserSession;
import com.etl.entity.EtlBusinessDbInfo;
import com.etl.entity.EtlTaskLog;
import com.etl.entity.EtlTaskMain;
import com.etl.exception.GlobalException;
import com.etl.mapper.EtlTaskLogMapper;
import com.etl.mapper.EtlTaskMainMapper;
import com.etl.service.KettleService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.pentaho.di.core.KettleEnvironment;
import org.pentaho.di.core.database.DatabaseMeta;
import org.pentaho.di.core.logging.LogLevel;
import org.pentaho.di.job.Job;
import org.pentaho.di.job.JobMeta;
import org.pentaho.di.repository.RepositoryDirectoryInterface;
import org.pentaho.di.repository.kdr.KettleDatabaseRepository;
import org.pentaho.di.repository.kdr.KettleDatabaseRepositoryMeta;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.regex.Pattern;

@Slf4j
@Service
public class KettleServiceImpl implements KettleService {

    @Value("${kettle.db-name}")
    private String kettleDbName;

    @Value("${kettle.db-url}")
    private String kettleDbUrl;

    @Value("${kettle.db-user}")
    private String kettleDbUser;

    @Value("${kettle.db-psd}")
    private String kettleDbPsd;

    @Value("${kettle.db-port}")
    private String kettleDbPort;

    @Value("${kettle.password}")
    private String kettlePassword;


    private DatabaseMeta databaseMeta;

    private KettleDatabaseRepository repository;


    private static Pattern dateRex = Pattern.compile("\\d{4}/\\d{2}/\\d{2} \\d{2}:\\d{2}:\\d{2}");



    @Resource
    private EtlTaskMainMapper etlTaskMainMapper;


    @Resource
    private EtlTaskIncrementMapper etlTaskIncrementMapper;


    @Resource
    private EtlTaskLogMapper etlTaskLogMapper;



    public KettleServiceImpl() {
        //初始化kettle环境
        try {
            KettleEnvironment.init();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    /**
     * 【私有】执行任务方法
     * batch传null 表示为单点
     */
    public int executeTask(Map<String,String> jobVariableMap,  String kettleHome,String jobName
            ,EtlBusinessDbInfo sourceDbInfo, EtlBusinessDbInfo tgtDbInfo, List<Long> kids,String logType) {
        int errors = 0 ;
        try {

            //数据库连接元对象
            if (databaseMeta == null) {
                log.info("数据库连接初始化…………………………");
                //ETL数据库,mysql
                databaseMeta = new DatabaseMeta("数仓数据库", "MYSQL", "Native", kettleDbUrl,  kettleDbName, kettleDbPort, kettleDbUser, kettleDbPsd);

                // 连接测试
                log.info(databaseMeta.testConnection());
                if (databaseMeta.testConnection().startsWith("正确") || databaseMeta.testConnection().contains("successful")) {
                    log.info("kettle资源库连接成功");
                } else {
                    log.error(ResultEnum.KETTLE_DB_ERROR.getErrMsg());
                    log.error(databaseMeta.testConnection());
                    throw new GlobalException(ResultEnum.KETTLE_DB_ERROR);
                }
            }
            //资源库元对象
            KettleDatabaseRepositoryMeta repInfo = new KettleDatabaseRepositoryMeta();
            repInfo.setConnection(databaseMeta);
            //资源库
            if (repository == null) {
                repository = new KettleDatabaseRepository();
                repository.init(repInfo);
                repository.connect("admin", kettlePassword);
            }
            // 获取目录
            RepositoryDirectoryInterface directory = repository.findDirectory("/"+kettleHome+"/");
            // 选择资源库中的作业
            JobMeta jobMeta = repository.loadJob(jobName, directory, null, "1.0");
            // 加载作业
            Job job = new Job(repository, jobMeta);


            //遍历参数，设置变量
            jobVariableMap.forEach((key, value)->{
                job.setVariable(key, value);
            });



            //拼装当前etl工具数据库连接
            job.setVariable("local_db_url", String.format(CommonConstant.MYSQL_URL, kettleDbUrl, kettleDbPort, kettleDbName));
            job.setVariable("local_db_driver", CommonConstant.MYSQL_DRIVER);
            job.setVariable("local_db_user", kettleDbUser);
            job.setVariable("local_db_psd", kettleDbPsd);
            job.setLogLevel(LogLevel.DEBUG);


            //拼装源数据库连接
            Map<String, String> sourceDbMap = CommonUtil.pingDBUrlAndDriver(sourceDbInfo);
            job.setVariable("source_db_url", sourceDbMap.get("url"));
            job.setVariable("source_db_driver", sourceDbMap.get("driver"));
            job.setVariable("source_db_user", sourceDbInfo.getUsername());
            job.setVariable("source_db_psd", AESUtil.decrypt(sourceDbInfo.getPassword(), AESUtil.password));
            job.setVariable("source_db_schema", sourceDbInfo.getDbSchema());

            //拼装目标数据库连接
            Map<String, String> tgtDbInfoMap = CommonUtil.pingDBUrlAndDriver(tgtDbInfo);
            job.setVariable("target_db_url", tgtDbInfoMap.get("url") );
            job.setVariable("target_db_driver", tgtDbInfoMap.get("driver"));
            job.setVariable("target_db_user", tgtDbInfo.getUsername());
            job.setVariable("target_db_psd", AESUtil.decrypt(tgtDbInfo.getPassword(), AESUtil.password));
            job.setVariable("target_db_schema", tgtDbInfo.getDbSchema());

            //设置目前数据库其它参数
            //判断数据库类型
            if( CommonConstant.ORACLE_NAME.equalsIgnoreCase(tgtDbInfo.getDbType())){
                job.setVariable("target_db_name","/"+tgtDbInfo.getDbName());
            }else{
                job.setVariable("target_db_name", tgtDbInfo.getDbName());
            }
            job.setVariable("target_db_type", tgtDbInfo.getDbType().toUpperCase());
            job.setVariable("target_db_port", tgtDbInfo.getDbPort());
            job.setVariable("target_db_ipv", tgtDbInfo.getDbIp());

            // 执行作业
            job.start();

            // 等待作业执行完毕
            job.waitUntilFinished();

            //记录操作日志关联
            this.saveEtlLog(job,kids,logType);

//            if (job.getErrors() > 0) {
//                throw new GlobalException(500, "作业执行出错");
//            }
            errors = job.getErrors();
        }catch (Exception e){
            log.error("KettleException:"+e);
        }

        return errors;

    }


    public void  saveEtlLog(Job job,List<Long> kids,String logType){

        if("main".equals(logType)){
            //记录日志
            QueryWrapper<EtlTaskMain> queryWrapper = new QueryWrapper<>();
            queryWrapper.eq("flag", "1").last(String.format("and kid in (%s)", StringUtils.join(kids, ",")));
            List<EtlTaskMain> etlTaskMainList = etlTaskMainMapper.selectList(queryWrapper);

            for (EtlTaskMain item : etlTaskMainList) {
                int flag = 0;
                if (item.getExecFlag() == 3) {
                    //抽取成功
                    flag = 1;
                }
                EtlTaskLog etlTaskLog = new EtlTaskLog();
                if (UserSession.getUser() != null) {
                    etlTaskLog.setAccountName(UserSession.getUser().getAccountName());
                } else {
                    etlTaskLog.setAccountName("sysadmin");
                }
                etlTaskLog.setKid(item.getKid()).setBatchId(job.getBatchId()).setStatus(flag).setLogType(logType).setCreateDate(new Date());
                etlTaskLogMapper.insert(etlTaskLog);


            }
        }else if("increment".equals(logType)){
            //记录日志
            QueryWrapper<EtlTaskIncrement> queryWrapper = new QueryWrapper<>();
            queryWrapper.eq("flag", "1").last(String.format("and kid in (%s)", StringUtils.join(kids, ",")));
            List<EtlTaskIncrement> etlTaskIncrementList = etlTaskIncrementMapper.selectList(queryWrapper);

            for (EtlTaskIncrement item : etlTaskIncrementList) {
                int flag = 0;
                if (item.getExecFlag() == 3) {
                    //抽取成功
                    flag = 1;
                }
                EtlTaskLog etlTaskLog = new EtlTaskLog();
                if (UserSession.getUser() != null) {
                    etlTaskLog.setAccountName(UserSession.getUser().getAccountName());
                } else {
                    etlTaskLog.setAccountName("sysadmin");
                }
                etlTaskLog.setKid(item.getKid()).setBatchId(job.getBatchId()).setStatus(flag).setLogType(logType).setCreateDate(new Date());
                etlTaskLogMapper.insert(etlTaskLog);
            }

        }




    }



}
